package org.sxp.modules.dataworks.controller;

import java.util.List;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.xxl.job.admin.controller.JobInfoController;
import com.xxl.job.admin.core.model.XxlJobInfo;
import com.xxl.job.admin.core.model.XxlJobLog;
import com.xxl.job.admin.dao.XxlJobGroupDao;
import com.xxl.job.admin.dao.XxlJobInfoDao;
import com.xxl.job.admin.dao.XxlJobLogDao;
import com.xxl.job.admin.service.XxlJobService;
import com.xxl.job.core.biz.model.ReturnT;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.sxp.common.dto.datax.DataXJobRoot;
import org.sxp.common.exception.BusinessException;
import org.sxp.common.utils.*;
import org.sxp.common.utils.datax.DataxJsonHelper;
import org.sxp.modules.dataworks.entity.SyncDO;
import org.sxp.modules.dataworks.service.SyncService;
import org.sxp.modules.dataworks.dto.SyncDTO;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.springframework.web.bind.annotation.*;
import org.sxp.common.annotation.SysLog;
import org.sxp.common.web.WrapperResponse;
import org.sxp.modules.dataworks.utils.XxlJobLogCaptureUtil;
import org.sxp.modules.sys.controller.AbstractController;
import org.apache.shiro.authz.annotation.RequiresPermissions;

import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;

/**
 * 
 * @author 沈兴平
 * @date 2021/12/30
 */
@Controller
@RequestMapping(value = "/dataworks/sync")
@Api(description="Sync", tags = "")
@SysLog(modules = "dataworks", value = "")
public class SyncController extends AbstractController{
    @Autowired
    private SyncService syncService;

    @Resource
    public XxlJobLogDao xxlJobLogDao;

    @Resource
    private XxlJobGroupDao xxlJobGroupDao;
    @Resource
    private XxlJobInfoDao xxlJobInfoDao;
    @Resource
    private JobInfoController jobInfoController;
    @Resource
    private XxlJobService xxlJobService;

    @RequestMapping("/index")
    public String module(HttpServletRequest request, Model model){
        jobInfoController.index(request, model, -1);
        return "modules/dataworks/sync.html";
    }
    /**
     * 列表
     */
    @ApiImplicitParams({
            @ApiImplicitParam(name="pageNum", value = "当前页码", dataType="int", paramType = "header"),
            @ApiImplicitParam(name="pageSize",value="每页条目",dataType="int", paramType = "header"),
            @ApiImplicitParam(name="orderField",value="排序字段",dataType="string", paramType = "header"),
            @ApiImplicitParam(name="orderType",value="排序类型",dataType="string", paramType = "header", example = "asc或者desc")
    })
    @RequestMapping(value = "/list",method = {RequestMethod.POST})
    @ApiOperation(value = "查询列表")
    @RequiresPermissions("dataworks:sync:list")
    @ResponseBody
    public WrapperResponse<List<SyncDTO>> list(SyncDTO params) throws Exception{
        //请求头获取分页参数
        PageInfo pageInfo = buildPageInfo();
        pageInfo.setOrderField(Query.fieldToColumn(SyncDO.class, AeyeReflectionUtil.getFieldName(SyncDO::getCreateTime)));
        pageInfo.setOrderType(Constant.DESC);
        IPage<SyncDO> page = syncService.page(
                new Query<SyncDO>().getPage(pageInfo),
                new LambdaQueryWrapper<SyncDO>()
                .eq(SyncDO::getSpaceId, params.getSpaceId())
                .like(StrUtil.isNotBlank(params.getName()), SyncDO::getName, params.getName())
        );
        for(SyncDO syncDO : page.getRecords()){
            if(syncDO.getJobinfoId() != null){
                XxlJobInfo jobInfo = xxlJobInfoDao.loadById(syncDO.getJobinfoId());
                if(jobInfo != null){
                    syncDO.setJobinfoName(jobInfo.getName());
                }
            }
        }
        return (WrapperResponse)WrapperResponse.success(new PageResult(page));
    }


    /**
     * 信息
     */
    @RequestMapping(value = "/info/{id}", method = {RequestMethod.GET})
    @ApiOperation(value = "查询")
    @RequiresPermissions("dataworks:sync:info")
    @ResponseBody
    public WrapperResponse<SyncDTO> info(@PathVariable String id) throws Exception{
        SyncDTO sync = XPBeanUtils.copyBean(syncService.getById(id), SyncDTO.class);
        return (WrapperResponse)WrapperResponse.success(DataxJsonHelper.backBuildDataXJob(sync));
    }

    /**
     * 工作保存与修改
     */
    @RequestMapping(value = "/syncSaveOrUpdate",method = {RequestMethod.POST})
    @ApiOperation(value = "保存")
    @SysLog("保存")
    @RequiresPermissions("dataworks:sync:save")
    @ResponseBody
    public WrapperResponse<SyncDTO> save(String id,String name, String spaceId) throws Exception{
        SyncDO syncDO = new SyncDO();
        syncDO.setName(name);
        syncDO.setSpaceId(spaceId);
        syncDO.setId(id);
        if(StrUtil.isNotBlank(syncDO.getId())){
            syncService.updateById(syncDO);
        }else{
            syncService.save(syncDO);
        }

        return (WrapperResponse)WrapperResponse.success(syncDO.getId());
    }

    /**
     * 修改
     */
    @RequestMapping(value = "/update",method = {RequestMethod.POST})
    @ApiOperation(value = "修改")
    @SysLog("修改")
    @RequiresPermissions("dataworks:sync:update")
    @ResponseBody
    public WrapperResponse<SyncDTO> update(@RequestBody SyncDTO sync) throws Exception{
        if(sync.getSyncRate() == null || sync.getSyncRate() == 0){
            sync.setSyncRate(-1);
        }
        if(sync.getSyncRecord() == null || sync.getSyncRecord() == 0){
            sync.setSyncRecord(-1);
        }
        DataXJobRoot jobRoot = DataxJsonHelper.buildDataXJob(sync);
        // 回写保存到同步表
        sync.setDataxJson(JSON.toJSONString(jobRoot));
        syncService.updateById(SyncDO.copyBean(sync));
        return (WrapperResponse)WrapperResponse.success(null);
    }

    /**
     * 批量删除
     */
    @RequestMapping(value = "/deleteBatch",method = {RequestMethod.POST})
    @ApiOperation(value = "批量删除")
    @SysLog("批量删除")
    @RequiresPermissions("dataworks:sync:deleteBatch")
    @ResponseBody
    public WrapperResponse deleteBatch(@RequestBody String[] ids) throws Exception{
        for(String id : ids){
            SyncDO syncDO = syncService.getById(id);
            boolean boo = syncService.removeById(id);
            if(boo && syncDO.getJobinfoId() != null){
                xxlJobService.remove(syncDO.getJobinfoId());
            }
        }
        return WrapperResponse.success(null);
    }

    /**
     * 绑定调度
     */
    @RequestMapping(value = "/bindJob", method = {RequestMethod.POST})
    @ApiOperation(value = "绑定调度")
    @ResponseBody
    public WrapperResponse bindJob(@RequestBody XxlJobInfo jobInfo, @RequestParam String id) throws Exception{
        SyncDO entity = syncService.getById(id);
        ReturnT<String> r = null;
        if(jobInfo.getId() > 0){
            ReturnT<XxlJobInfo> oldJobInfo = jobInfoController.loadById(jobInfo.getId());

            r = infoController.update(jobInfo);
            r.setContent(String.valueOf(jobInfo.getId()));
            if(r.getCode() == ReturnT.SUCCESS_CODE && oldJobInfo.getContent().getTriggerStatus() != jobInfo.getTriggerStatus()){
                // 绑定调度的运行状态不等于老数据
                if(jobInfo.getTriggerStatus() == 1){
                    jobInfoController.start(jobInfo.getId());
                }else{
                    jobInfoController.pause(jobInfo.getId());
                }
            }
        }else{
            r = infoController.add(jobInfo);
            if(r.getCode() == ReturnT.SUCCESS_CODE){
                entity.setJobinfoId(Integer.parseInt(r.getContent()));
                syncService.updateById(entity);

                // 绑定调度的运行状态不等于老数据
                if(jobInfo.getTriggerStatus() == 1){
                    jobInfoController.start(jobInfo.getId());
                }
            }
        }
        return r.getCode() == ReturnT.SUCCESS_CODE ? WrapperResponse.success(r.getContent()) : WrapperResponse.fail(r.getMsg(), r.getContent());
    }

    /**
     * 查询是否存在进行中调度
     */
    @RequestMapping(value = "/findJobActiveLog",method = {RequestMethod.GET})
    @ApiOperation(value = "查询是否存在进行中调度")
    @SysLog("查询是否存在进行中调度")
    @ResponseBody
    public WrapperResponse findJobActiveLog(Integer xxlJobId) throws Exception{
        List<XxlJobLog> activeLog = xxlJobLogDao.findJobActiveLog(xxlJobId);
        return WrapperResponse.success(activeLog.size() > 0 ? activeLog.get(0) : null);
    }

    @Autowired
    private JobInfoController infoController;
    /**
     * 执行
     */
    @RequestMapping(value = "/executeJob",method = {RequestMethod.POST})
    @ApiOperation(value = "执行")
    @SysLog("执行")
    @RequiresPermissions("dataworks:sync:execute")
    @ResponseBody
    public WrapperResponse executeJob(@RequestBody SyncDTO sync) throws Exception{
        List<XxlJobLog> activeLog = xxlJobLogDao.findJobActiveLog(sync.getJobinfoId());
        if(activeLog != null && activeLog.size() > 0){
            return WrapperResponse.fail("工作绑定的调度正在执行任务，请结束后再执行!!!", null);
        }
        // 先做保存顺便把改校验的校验了
        update(sync);

        ReturnT<XxlJobInfo> returnT = infoController.loadById(sync.getJobinfoId());
        if(returnT.getCode() == ReturnT.SUCCESS_CODE && returnT.getContent() != null){
            XxlJobInfo jobInfo = returnT.getContent();
            try {
                String snapshotId = IdUtil.simpleUUID();
                XxlJobLogCaptureUtil.jobLogContext.set(snapshotId);
                ReturnT<String> triggerReturnT = infoController.triggerJob(sync.getJobinfoId(), jobInfo.getExecutorParam(), null);
                if (triggerReturnT.getCode() == ReturnT.SUCCESS_CODE) {
                    int index = 5;
                    while(index > 0){
                        XxlJobLog jobLog = xxlJobLogDao.loadBySnapshotId(snapshotId);
                        if(jobLog != null){
                            return WrapperResponse.success(jobLog.getId());
                        }
                        Thread.sleep(1000);
                        index--;
                    }
                } else {
                    throw new BusinessException("调度执行失败:" + triggerReturnT.getMsg());
                }
            }catch (Exception e){
                throw e;
            }finally {
                XxlJobLogCaptureUtil.jobLogContext.remove();
            }
        }else{
            return WrapperResponse.fail("调度任务不存在，可能被移除，请重新配置。", null);
        }
        return WrapperResponse.success(null);
    }
}
